import os
import pandas as pd
import numpy as np
import logging
import time
import akshare as ak
from qtorch.core.interfaces import IDataProvider
from qtorch.core.datafetcher import DataFetcher
from qtorch.core.exceptions import FinancialDataException
from qtorch.utils.validators import ValidationError
from typing import Optional, Dict, Any, List, Tuple
from qtorch.tool import retry
from qtorch.data.aksharestockerfetcher import AKShareStockFetcher

# 设置基本日志配置
logging.basicConfig(level=logging.INFO)

class MultiAssetFetcher(DataFetcher, IDataProvider):
    """
    多资产数据获取器，支持股票、期货、期权等多种金融工具
    """
    
    def __init__(self, cache_dir="data_cache", cache_ttl=86400):
        """
        初始化多资产数据获取器
        
        参数:
            cache_dir: 缓存目录
            cache_ttl: 缓存过期时间（秒）
        """
        super().__init__()
        self.cache_dir = cache_dir
        self.cache_ttl = cache_ttl
        os.makedirs(self.cache_dir, exist_ok=True)
        
        # 股票数据获取器
        self.stock_fetcher = AKShareStockFetcher(cache_dir=f"{cache_dir}/stock", cache_ttl=cache_ttl)
        
        # 资产类型映射 - 用于标识不同的资产类型
        self.asset_type_map = {
            # 股票
            'stock': lambda symbol: symbol.startswith(('sh', 'sz', '00', '60', '30', '68')) and not symbol.startswith(('sh51', 'sz15')),
            
            # ETF
            'etf': lambda symbol: symbol.startswith(('sh51', 'sz15')) or '.ETF' in symbol,
            
            # 期货
            'futures': lambda symbol: symbol.endswith(('.CFX', '.SHF', '.DCE', '.CZC', '.INE')),
            
            # 期权
            'option': lambda symbol: symbol.endswith(('.opt')),
            
            # 债券
            'bond': lambda symbol: symbol.startswith(('sh12', 'sz12')),
            
            # 加密货币
            'crypto': lambda symbol: symbol.endswith(('.BTC', '.ETH')),
            
            # 外汇
            'forex': lambda symbol: symbol.endswith(('.FOREX')) or '/' in symbol or symbol in ['EURUSD', 'USDJPY', 'GBPUSD', 'AUDUSD', 'USDCAD', 'USDCHF', 'NZDUSD']
        }
        # 不同类型资产的数据字段映射
        self.field_mappings = {
            'stock': {
                '日期': 'date',
                '开盘': 'open',
                '收盘': 'close',
                '最高': 'high',
                '最低': 'low',
                '成交量': 'volume'
            },
            'etf': {
                '日期': 'date',
                '开盘': 'open',
                '收盘': 'close',
                '最高': 'high',
                '最低': 'low',
                '成交量': 'volume',
                '成交额': 'amount',
                '涨跌幅': 'change_pct',
                '净值': 'nav',
                '溢价率': 'premium_discount'
            },
            'futures': {
                '日期': 'date',
                '开盘价': 'open',
                '收盘价': 'close',
                '最高价': 'high',
                '最低价': 'low',
                '成交量': 'volume',
                '持仓量': 'open_interest',
                '交易金额': 'turnover'
            },
            'option': {
                '日期': 'date',
                '开盘价': 'open',
                '收盘价': 'close',
                '最高价': 'high',
                '最低价': 'low',
                '成交量': 'volume',
                '持仓量': 'open_interest',
                '隐含波动率': 'implied_volatility',
                '行权价': 'strike_price',
                '到期日': 'expiry_date'
            },
            'forex': {
                '日期': 'date',
                '开盘价': 'open',
                '收盘价': 'close',
                '最高价': 'high',
                '最低价': 'low',
                '成交量': 'volume',
                '买入价': 'bid',
                '卖出价': 'ask',
                '点差': 'spread'
            }
        }
    
    def _get_asset_type(self, symbol: str) -> str:
        """
        根据符号确定资产类型
        
        参数:
            symbol: 资产代码
            
        返回:
            资产类型: 'stock', 'etf', 'futures', 'option', 'bond', 'crypto', 'forex'
        """
        for asset_type, check_func in self.asset_type_map.items():
            if check_func(symbol):
                return asset_type
        return 'stock'  # 默认为股票类型
    
    def _get_cache_path(self, symbol: str, asset_type: str = None) -> str:
        """
        获取缓存路径
        
        参数:
            symbol: 资产代码
            asset_type: 资产类型
            
        返回:
            缓存文件路径
        """
        if asset_type is None:
            asset_type = self._get_asset_type(symbol)
            
        asset_cache_dir = os.path.join(self.cache_dir, asset_type)
        os.makedirs(asset_cache_dir, exist_ok=True)
        return os.path.join(asset_cache_dir, f"{symbol}.parquet")
    
    def _load_from_cache(self, symbol: str, asset_type: str = None) -> Optional[pd.DataFrame]:
        """
        从缓存加载数据
        
        参数:
            symbol: 资产代码
            asset_type: 资产类型
            
        返回:
            数据DataFrame或None
        """
        cache_path = self._get_cache_path(symbol, asset_type)
        if os.path.exists(cache_path):
            file_mtime = os.path.getmtime(cache_path)
            if time.time() - file_mtime < self.cache_ttl:
                try:
                    return pd.read_parquet(cache_path)
                except Exception as e:
                    logging.warning(f"缓存加载失败: {e}")
        return None
    
    def _save_to_cache(self, symbol: str, data: pd.DataFrame, asset_type: str = None):
        """
        保存数据到缓存
        
        参数:
            symbol: 资产代码
            data: 数据DataFrame
            asset_type: 资产类型
        """
        try:
            cache_path = self._get_cache_path(symbol, asset_type)
            data.to_parquet(cache_path)
        except Exception as e:
            logging.error(f"缓存保存失败: {e}")
    
    @retry(max_retries=3, delay=5)
    def get_market_data(self, symbol: str, start_date: str, end_date: str, asset_type: str = None) -> pd.DataFrame:
        """
        获取市场数据
        
        参数:
            symbol: 资产代码
            start_date: 开始日期
            end_date: 结束日期
            asset_type: 资产类型，如不提供则自动检测
            
        返回:
            市场数据DataFrame
        """
        try:
            # 检查日期格式
            pd.to_datetime(start_date)
            pd.to_datetime(end_date)
            
            # 确定资产类型
            if asset_type is None:
                asset_type = self._get_asset_type(symbol)
            
            logging.info(f"获取{asset_type}类型资产 {symbol} 的数据")
            
            # 获取数据
            data = self._fetch_data_by_type(symbol, start_date, end_date, asset_type)
            
            # 数据验证
            if data.empty:
                logging.warning(f"未获取到{symbol}的有效数据")
                return pd.DataFrame()
                
            # 数据预处理
            processed_data = self.preprocess_data(data, asset_type)
            
            # 添加标识
            processed_data['asset_type'] = asset_type
            processed_data['is_market'] = True
            
            return processed_data
            
        except Exception as e:
            logging.error(f"获取{symbol}数据失败: {str(e)}")
            raise FinancialDataException(f"市场数据获取失败: {symbol}") from e
    
    def _fetch_data_by_type(self, symbol: str, start_date: str, end_date: str, asset_type: str) -> pd.DataFrame:
        """
        根据资产类型获取数据
        
        参数:
            symbol: 资产代码
            start_date: 开始日期
            end_date: 结束日期
            asset_type: 资产类型
            
        返回:
            市场数据DataFrame
        """
        # 优先从缓存加载
        cached_data = self._load_from_cache(symbol, asset_type)
        if cached_data is not None:
            return cached_data[(cached_data.index >= start_date) & (cached_data.index <= end_date)]
        
        # 根据资产类型调用不同的数据获取方法
        if asset_type == 'stock':
            return self._fetch_stock_data(symbol, start_date, end_date)
        elif asset_type == 'etf':
            return self._fetch_etf_data(symbol, start_date, end_date)
        elif asset_type == 'futures':
            return self._fetch_futures_data(symbol, start_date, end_date)
        elif asset_type == 'option':
            return self._fetch_option_data(symbol, start_date, end_date)
        elif asset_type == 'bond':
            return self._fetch_bond_data(symbol, start_date, end_date)
        elif asset_type == 'crypto':
            return self._fetch_crypto_data(symbol, start_date, end_date)
        elif asset_type == 'forex':
            return self._fetch_forex_data(symbol, start_date, end_date)
        else:
            logging.warning(f"未知资产类型: {asset_type}，默认当作股票处理")
            return self._fetch_stock_data(symbol, start_date, end_date)
    
    def _fetch_stock_data(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
        """获取股票数据"""
        try:
            # 直接调用现有的股票数据获取器
            data = self.stock_fetcher.get_market_data(symbol, start_date, end_date)
            self._save_to_cache(symbol, data, 'stock')
            return data
        except Exception as e:
            logging.error(f"获取股票数据失败: {str(e)}")
            return pd.DataFrame()
    
    def _fetch_futures_data(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
        """获取期货数据"""
        try:
            logging.info(f"获取期货数据: {symbol}")
            
            # 调用AKShare获取期货数据
            if symbol.endswith('.SHF'):  # 上海期货交易所
                exchange = "shfe"
                symbol_code = symbol.split('.')[0]
            elif symbol.endswith('.DCE'):  # 大连商品交易所
                exchange = "dce"
                symbol_code = symbol.split('.')[0]
            elif symbol.endswith('.CZC'):  # 郑州商品交易所
                exchange = "czce"
                symbol_code = symbol.split('.')[0]
            elif symbol.endswith('.CFX'):  # 中金所
                exchange = "cffex"
                symbol_code = symbol.split('.')[0]
            else:
                exchange = "shfe"  # 默认上期所
                symbol_code = symbol
            
            # 调用akshare获取期货历史数据
            try:
                futures_df = ak.futures_main_sina(symbol=symbol_code, start_date=start_date, end_date=end_date)
            except:
                # 如果主力合约数据获取失败，尝试获取指定合约数据
                futures_df = ak.futures_zh_daily_sina(symbol=symbol_code)
            
            if futures_df.empty:
                return pd.DataFrame()
            
            # 重命名列
            field_mapping = self.field_mappings.get('futures', {})
            futures_df.rename(columns={k: v for k, v in field_mapping.items() if k in futures_df.columns}, inplace=True)
            
            # 转换日期格式
            futures_df['date'] = pd.to_datetime(futures_df['date'])
            futures_df.set_index('date', inplace=True)
            
            # 过滤日期范围
            start_dt = pd.to_datetime(start_date)
            end_dt = pd.to_datetime(end_date)
            futures_df = futures_df[(futures_df.index >= start_dt) & (futures_df.index <= end_dt)]
            
            # 确保所需列存在
            required_cols = ['open', 'high', 'low', 'close', 'volume']
            for col in required_cols:
                if col not in futures_df.columns:
                    futures_df[col] = 0.0
            
            # 数据类型转换
            for col in required_cols:
                futures_df[col] = pd.to_numeric(futures_df[col], errors='coerce')
            
            # 缓存处理
            self._save_to_cache(symbol, futures_df, 'futures')
            
            return futures_df
            
        except Exception as e:
            logging.error(f"获取期货数据失败: {str(e)}")
            return pd.DataFrame()
    
    def _fetch_option_data(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
        """获取期权数据"""
        try:
            logging.info(f"获取期权数据: {symbol}")
            
            # 期权数据获取逻辑
            # 目前AKShare对期权支持有限，这里提供示例
            option_df = None
            
            try:
                # 尝试从AKShare获取期权数据
                # 根据期权类型可能需要调用不同的API
                if '50ETF' in symbol or '510050' in symbol:
                    # 上证50ETF期权
                    option_df = ak.option_sse_list_sina()
                elif '300ETF' in symbol or '510300' in symbol:
                    # 沪深300ETF期权
                    option_df = ak.option_sse_list_sina()
                    
                # 如果找到合约代码，获取历史数据
                if option_df is not None and not option_df.empty:
                    target_option = option_df[option_df['symbol'].str.contains(symbol)]
                    if not target_option.empty:
                        option_code = target_option.iloc[0]['symbol']
                        option_df = ak.option_sse_daily_sina(symbol=option_code)
            except Exception as option_e:
                logging.error(f"AKShare期权数据获取失败: {str(option_e)}")
                option_df = None
                
            # 如果AKShare获取失败，创建模拟数据以满足开发需要
            if option_df is None or option_df.empty:
                logging.warning(f"无法获取{symbol}的期权数据，创建模拟数据")
                
                # 创建日期序列
                date_range = pd.date_range(start=start_date, end=end_date, freq='B')
                
                # 创建模拟数据
                data = {
                    'date': date_range,
                    'open': np.random.normal(100, 5, len(date_range)),
                    'high': np.random.normal(105, 5, len(date_range)),
                    'low': np.random.normal(95, 5, len(date_range)),
                    'close': np.random.normal(102, 5, len(date_range)),
                    'volume': np.random.randint(1000, 10000, len(date_range)),
                    'open_interest': np.random.randint(5000, 15000, len(date_range)),
                    'implied_volatility': np.random.normal(0.2, 0.05, len(date_range))
                }
                
                option_df = pd.DataFrame(data)
                option_df['date'] = pd.to_datetime(option_df['date'])
                option_df.set_index('date', inplace=True)
                
                # 添加期权特有字段
                option_df['strike_price'] = 100.0  # 模拟行权价
                option_df['expiry_date'] = pd.Timestamp(end_date)  # 模拟到期日
                
            else:
                # 标准化实际期权数据
                field_mapping = self.field_mappings.get('option', {})
                option_df.rename(columns={k: v for k, v in field_mapping.items() if k in option_df.columns}, inplace=True)
                
                # 确保日期是索引
                if 'date' in option_df.columns:
                    option_df['date'] = pd.to_datetime(option_df['date'])
                    option_df.set_index('date', inplace=True)
                
                # 过滤日期范围
                start_dt = pd.to_datetime(start_date)
                end_dt = pd.to_datetime(end_date)
                option_df = option_df[(option_df.index >= start_dt) & (option_df.index <= end_dt)]
            
            # 确保所需列存在
            required_cols = ['open', 'high', 'low', 'close', 'volume']
            for col in required_cols:
                if col not in option_df.columns:
                    option_df[col] = 0.0
            
            # 缓存处理
            self._save_to_cache(symbol, option_df, 'option')
            
            return option_df
            
        except Exception as e:
            logging.error(f"获取期权数据失败: {str(e)}")
            return pd.DataFrame()
    
    def _fetch_bond_data(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
        """获取债券数据"""
        try:
            logging.info(f"获取债券数据: {symbol}")
            
            # 尝试从AKShare获取债券数据
            try:
                bond_df = ak.bond_zh_hs_daily(symbol=symbol)
            except:
                # 创建模拟数据
                date_range = pd.date_range(start=start_date, end=end_date, freq='B')
                data = {
                    'date': date_range,
                    'open': np.random.normal(100, 0.1, len(date_range)),
                    'high': np.random.normal(100.1, 0.1, len(date_range)),
                    'low': np.random.normal(99.9, 0.1, len(date_range)),
                    'close': np.random.normal(100, 0.1, len(date_range)),
                    'volume': np.random.randint(1000, 5000, len(date_range))
                }
                bond_df = pd.DataFrame(data)
            
            # 标准化数据
            bond_df['date'] = pd.to_datetime(bond_df['date'])
            bond_df.set_index('date', inplace=True)
            
            # 过滤日期范围
            start_dt = pd.to_datetime(start_date)
            end_dt = pd.to_datetime(end_date)
            bond_df = bond_df[(bond_df.index >= start_dt) & (bond_df.index <= end_dt)]
            
            # 确保所需列存在
            required_cols = ['open', 'high', 'low', 'close', 'volume']
            for col in required_cols:
                if col not in bond_df.columns:
                    bond_df[col] = 0.0
            
            # 缓存处理
            self._save_to_cache(symbol, bond_df, 'bond')
            
            return bond_df
            
        except Exception as e:
            logging.error(f"获取债券数据失败: {str(e)}")
            return pd.DataFrame()
    
    def _fetch_crypto_data(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
        """获取加密货币数据"""
        try:
            logging.info(f"获取加密货币数据: {symbol}")
            
            # 尝试从AKShare获取加密货币数据
            crypto_type = symbol.split('.')[1] if '.' in symbol else 'BTC'
            
            try:
                if crypto_type == 'BTC':
                    crypto_df = ak.crypto_hist(symbol="BTC-USD")
                elif crypto_type == 'ETH':
                    crypto_df = ak.crypto_hist(symbol="ETH-USD")
                else:
                    crypto_df = ak.crypto_hist(symbol=f"{crypto_type}-USD")
            except:
                # 创建模拟数据
                date_range = pd.date_range(start=start_date, end=end_date, freq='D')
                data = {
                    'date': date_range,
                    'open': np.random.normal(40000, 1000, len(date_range)) if crypto_type == 'BTC' else np.random.normal(2000, 100, len(date_range)),
                    'high': np.random.normal(41000, 1000, len(date_range)) if crypto_type == 'BTC' else np.random.normal(2050, 100, len(date_range)),
                    'low': np.random.normal(39000, 1000, len(date_range)) if crypto_type == 'BTC' else np.random.normal(1950, 100, len(date_range)),
                    'close': np.random.normal(40500, 1000, len(date_range)) if crypto_type == 'BTC' else np.random.normal(2025, 100, len(date_range)),
                    'volume': np.random.randint(1000, 10000, len(date_range))
                }
                crypto_df = pd.DataFrame(data)
            
            # 标准化数据
            crypto_df['date'] = pd.to_datetime(crypto_df['date'])
            crypto_df.set_index('date', inplace=True)
            
            # 过滤日期范围
            start_dt = pd.to_datetime(start_date)
            end_dt = pd.to_datetime(end_date)
            crypto_df = crypto_df[(crypto_df.index >= start_dt) & (crypto_df.index <= end_dt)]
            
            # 确保所需列存在
            required_cols = ['open', 'high', 'low', 'close', 'volume']
            for col in required_cols:
                if col not in crypto_df.columns:
                    crypto_df[col] = 0.0
            
            # 缓存处理
            self._save_to_cache(symbol, crypto_df, 'crypto')
            
            return crypto_df
            
        except Exception as e:
            logging.error(f"获取加密货币数据失败: {str(e)}")
            return pd.DataFrame()
    
    def _fetch_etf_data(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
        """获取ETF数据"""
        try:
            logging.info(f"获取ETF数据: {symbol}")
            
            # 尝试从AKShare获取ETF数据
            try:
                # 根据ETF代码获取数据
                if symbol.startswith('sh51'):
                    # 上交所ETF
                    etf_df = ak.fund_etf_hist_sina(symbol=symbol.replace('sh', ''))
                elif symbol.startswith('sz15'):
                    # 深交所ETF
                    etf_df = ak.fund_etf_hist_sina(symbol=symbol.replace('sz', ''))
                else:
                    # 尝试直接获取
                    etf_df = ak.fund_etf_hist_sina(symbol=symbol)
                    
                # 如果获取失败，尝试使用股票接口
                if etf_df is None or etf_df.empty:
                    etf_df = ak.stock_zh_a_hist(symbol=symbol, period="daily")
            except Exception as etf_e:
                logging.error(f"AKShare ETF数据获取失败: {str(etf_e)}")
                # 尝试使用股票接口作为备选
                try:
                    etf_df = ak.stock_zh_a_hist(symbol=symbol, period="daily")
                except:
                    # 创建模拟数据
                    date_range = pd.date_range(start=start_date, end=end_date, freq='B')
                    data = {
                        'date': date_range,
                        'open': np.random.normal(3, 0.1, len(date_range)),
                        'high': np.random.normal(3.1, 0.1, len(date_range)),
                        'low': np.random.normal(2.9, 0.1, len(date_range)),
                        'close': np.random.normal(3, 0.1, len(date_range)),
                        'volume': np.random.randint(10000, 50000, len(date_range)),
                        'amount': np.random.randint(30000, 150000, len(date_range))
                    }
                    etf_df = pd.DataFrame(data)
            
            # 标准化数据
            field_mapping = self.field_mappings.get('etf', {})
            etf_df.rename(columns={k: v for k, v in field_mapping.items() if k in etf_df.columns}, inplace=True)
            
            # 确保日期列存在并格式正确
            if 'date' in etf_df.columns:
                etf_df['date'] = pd.to_datetime(etf_df['date'])
                etf_df.set_index('date', inplace=True)
            
            # 过滤日期范围
            start_dt = pd.to_datetime(start_date)
            end_dt = pd.to_datetime(end_date)
            etf_df = etf_df[(etf_df.index >= start_dt) & (etf_df.index <= end_dt)]
            
            # 确保所需列存在
            required_cols = ['open', 'high', 'low', 'close', 'volume']
            for col in required_cols:
                if col not in etf_df.columns:
                    etf_df[col] = 0.0
            
            # 尝试获取ETF额外信息（如净值、溢价率等）
            try:
                # 获取ETF基本信息
                etf_info = ak.fund_etf_fund_info_em(fund=symbol.replace('sh', '').replace('sz', ''))
                
                # 如果成功获取，添加额外信息
                if etf_info is not None and not etf_info.empty:
                    # 提取最新净值
                    latest_nav = etf_info.iloc[0]['单位净值'] if '单位净值' in etf_info.columns else None
                    
                    if latest_nav is not None:
                        # 创建净值列（使用最新净值作为基准，根据价格变动估算历史净值）
                        latest_close = etf_df['close'].iloc[-1]
                        etf_df['nav'] = etf_df['close'] / latest_close * latest_nav
                        
                        # 计算溢价/折价率
                        etf_df['premium_discount'] = (etf_df['close'] / etf_df['nav'] - 1) * 100
            except Exception as info_e:
                logging.warning(f"获取ETF额外信息失败: {str(info_e)}")
            
            # 缓存处理
            self._save_to_cache(symbol, etf_df, 'etf')
            
            return etf_df
            
        except Exception as e:
            logging.error(f"获取ETF数据失败: {str(e)}")
            return pd.DataFrame()
    
    def _fetch_forex_data(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
        """获取外汇数据"""
        try:
            logging.info(f"获取外汇数据: {symbol}")
            
            # 处理外汇符号格式
            forex_symbol = symbol
            if '/' in symbol:
                # 如果是"EUR/USD"格式，转换为"EURUSD"格式
                forex_symbol = symbol.replace('/', '')
            elif symbol.endswith('.FOREX'):
                # 如果是"EURUSD.FOREX"格式，去掉后缀
                forex_symbol = symbol.split('.')[0]
            
            # 尝试从AKShare获取外汇数据
            try:
                # 使用AKShare的外汇历史数据接口
                if forex_symbol in ['EURUSD', 'USDJPY', 'GBPUSD', 'AUDUSD', 'USDCAD', 'USDCHF', 'NZDUSD']:
                    # 主要货币对
                    forex_df = ak.fx_spot_quote()
                    
                    # 如果只返回了当前报价，创建历史数据
                    if len(forex_df) <= 1:
                        raise ValueError("只获取到当前报价，需要创建历史数据")
                else:
                    # 尝试其他接口
                    forex_df = ak.currency_history(symbol=forex_symbol, start_date=start_date, end_date=end_date)
            except Exception as forex_e:
                logging.error(f"AKShare外汇数据获取失败: {str(forex_e)}")
                
                # 创建模拟数据
                date_range = pd.date_range(start=start_date, end=end_date, freq='D')
                
                # 根据不同货币对设置基准价格
                base_price = 1.0
                if forex_symbol == 'EURUSD':
                    base_price = 1.1
                elif forex_symbol == 'USDJPY':
                    base_price = 110.0
                elif forex_symbol == 'GBPUSD':
                    base_price = 1.3
                elif forex_symbol == 'AUDUSD':
                    base_price = 0.7
                elif forex_symbol == 'USDCAD':
                    base_price = 1.25
                elif forex_symbol == 'USDCHF':
                    base_price = 0.9
                elif forex_symbol == 'NZDUSD':
                    base_price = 0.65
                
                # 生成模拟价格数据
                volatility = base_price * 0.005  # 0.5%的波动率
                data = {
                    'date': date_range,
                    'open': np.random.normal(base_price, volatility, len(date_range)),
                    'high': np.random.normal(base_price * 1.002, volatility, len(date_range)),
                    'low': np.random.normal(base_price * 0.998, volatility, len(date_range)),
                    'close': np.random.normal(base_price, volatility, len(date_range)),
                    'volume': np.random.randint(1000, 10000, len(date_range))
                }
                
                # 添加买入价和卖出价
                spread_pips = 2  # 2个基点的点差
                spread_value = base_price * (spread_pips / 10000)
                data['bid'] = [price - spread_value/2 for price in data['close']]
                data['ask'] = [price + spread_value/2 for price in data['close']]
                data['spread'] = spread_value
                
                forex_df = pd.DataFrame(data)
            
            # 标准化数据
            field_mapping = self.field_mappings.get('forex', {})
            forex_df.rename(columns={k: v for k, v in field_mapping.items() if k in forex_df.columns}, inplace=True)
            
            # 确保日期列存在并格式正确
            if 'date' in forex_df.columns:
                forex_df['date'] = pd.to_datetime(forex_df['date'])
                forex_df.set_index('date', inplace=True)
            
            # 过滤日期范围
            start_dt = pd.to_datetime(start_date)
            end_dt = pd.to_datetime(end_date)
            forex_df = forex_df[(forex_df.index >= start_dt) & (forex_df.index <= end_dt)]
            
            # 确保所需列存在
            required_cols = ['open', 'high', 'low', 'close', 'volume']
            for col in required_cols:
                if col not in forex_df.columns:
                    forex_df[col] = 0.0
            
            # 缓存处理
            self._save_to_cache(symbol, forex_df, 'forex')
            
            return forex_df
            
        except Exception as e:
            logging.error(f"获取外汇数据失败: {str(e)}")
            return pd.DataFrame()
    
    def preprocess_data(self, raw_data: pd.DataFrame, asset_type: str = 'stock') -> pd.DataFrame:
        """
        预处理不同资产类型的数据
        
        参数:
            raw_data: 原始数据
            asset_type: 资产类型
            
        返回:
            处理后的数据
        """
        df = raw_data.copy()
        
        # 计算收益率
        df["returns"] = df["close"].pct_change()
        
        # 计算波动率
        df["volatility"] = df["returns"].rolling(5).std()
        
        # 根据不同资产类型添加特定字段
        if asset_type == 'futures':
            # 期货特有处理 - 计算基差（如果有现货价格）
            if 'spot_price' in df.columns:
                df['basis'] = df['close'] - df['spot_price']
                df['basis_percent'] = df['basis'] / df['spot_price']
            
            # 加入持仓量变化
            if 'open_interest' in df.columns:
                df['oi_change'] = df['open_interest'].pct_change()
                
        elif asset_type == 'option':
            # 期权特有处理
            if all(x in df.columns for x in ['close', 'strike_price']):
                # 计算行权价值
                if 'option_type' in df.columns:
                    df['intrinsic_value'] = df.apply(
                        lambda row: max(row['close'] - row['strike_price'], 0)
                                   if row['option_type'] == 'call'
                                   else max(row['strike_price'] - row['close'], 0),
                        axis=1
                    )
                    
                    # 如果有隐含波动率，计算时间价值
                    if 'implied_volatility' in df.columns:
                        df['time_value'] = df['close'] - df['intrinsic_value']
        
        elif asset_type == 'etf':
            # ETF特有处理
            # 计算日内波动率
            df['intraday_volatility'] = (df['high'] - df['low']) / df['close']
            
            # 如果有净值数据，计算溢价/折价率
            if 'nav' in df.columns:
                df['premium_discount'] = (df['close'] / df['nav'] - 1) * 100
                
            # 计算跟踪误差（如果有基准指数）
            if 'benchmark_return' in df.columns:
                df['tracking_error'] = df['returns'] - df['benchmark_return']
                
        elif asset_type == 'forex':
            # 外汇特有处理
            # 计算点差比例
            if 'spread' in df.columns:
                df['spread_pct'] = df['spread'] / df['close'] * 10000  # 以基点表示
                
            # 计算买卖价差
            if all(x in df.columns for x in ['bid', 'ask']):
                df['bid_ask_spread'] = df['ask'] - df['bid']
                
            # 计算波动性指标
            df['atr'] = self._calculate_atr(df, 14)
        
        # 清理无效数据
        df.dropna(inplace=True)
        
        # 确保必要的列存在
        required_cols = ["open", "high", "low", "close", "volume", "returns", "volatility"]
        for col in required_cols:
            if col not in df.columns:
                df[col] = 0.0
        
        # 返回处理后的数据
        return df
        
    def _calculate_atr(self, df: pd.DataFrame, period: int = 14) -> pd.Series:
        """
        计算平均真实范围(ATR)
        
        参数:
            df: 包含价格数据的DataFrame
            period: 计算周期
            
        返回:
            pd.Series: ATR值
        """
        # 确保必要的列存在
        if not all(col in df.columns for col in ['high', 'low', 'close']):
            return pd.Series(0, index=df.index)
            
        # 计算真实范围(TR)
        high = df['high']
        low = df['low']
        close = df['close'].shift(1)
        
        # 处理第一行的NaN值
        close.iloc[0] = df['close'].iloc[0]
        
        # 计算三种TR值
        tr1 = high - low
        tr2 = (high - close).abs()
        tr3 = (low - close).abs()
        
        # 取最大值作为TR
        tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        
        # 计算ATR
        atr = tr.rolling(window=period).mean()
        
        return atr